package cn.tx.receive;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author Mxb
 * @version 1.0
 * @date 2020/10/13 22:19
 */
@Component
public class MessageReceive5 {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private BitMapBloomFilter bitMapBloomFilter;
    //消息的前缀
    private String MESSAGE = "message:";

    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue("queue"),
                    key = "error",
                    exchange = @Exchange(value = "directs"))
            //默认的直连交换机
    })
    public void receiveMessage(String content, Message message, Channel channel)
            throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        //消息投递ID
        System.out.println("消息投递ID:" + deliveryTag);
        String messageId = message.getMessageProperties().getMessageId();
        System.out.println("消息自定义ID:" + messageId);
        // 在布隆过滤器中就行判断
        if (bitMapBloomFilter.contains(messageId)) {
            System.out.println("这个消息被消费过，不能重复消费");
            try {
                //如果进入到这里面，说明这个消息之前被消费过，但是MQ 认为你没有消费，所以我们要签收这条消息
                channel.basicAck(deliveryTag, false);
                return;
            } catch (Exception e) {
                System.out.println(e);
            }
        }
        if (content.equals("123456")) {//如果消息内容为123456 就签收它
            /**
             * 参数说明:
             * deliveryTag 消息投递ID，要签收的投递ID 是多少
             82
             * multiple:是否批量签收
             */
            channel.basicAck(deliveryTag, false);
            System.out.println("消息签收成功");
            //消费成功之后放到布隆过滤器里面
            bitMapBloomFilter.add(messageId);
        } else {
            String count = redisTemplate.opsForValue().get(MESSAGE + messageId);
            if (count != null && Long.valueOf(count) >= 3) {
                channel.basicNack(deliveryTag, false, false);
                System.out.println("该消息消费3 失败，我们记录它，人工处理:" + content);
            } else {
                //如果不是123456 就拒绝签收
                /**
                 * 参数说明:
                 * deliveryTag 消息投递ID，要签收的投递ID 是多少
                 * multiple:是否批量签收
                 * requeue: true 代表拒绝签收并把消息重新放回到队列里面false 就直接拒绝
                 */
                //处理业务逻辑的处理[可能逻辑出现问题]
                channel.basicNack(deliveryTag, false, true);
                System.out.println("消息被拒绝签收");
                //因现被拒绝了，我们把消息ID 放到redis 里面
                redisTemplate.opsForValue().increment(MESSAGE + messageId);
            }
        }
        System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
    }
}
